package com.base.core.producer.task;

import com.base.core.producer.broker.RabbitBroker;
import com.base.core.producer.constant.BrokerMessageStatus;
import com.base.core.producer.entity.BrokerMessage;
import com.base.core.producer.service.MessageStoreService;
import com.base.core.task.annotation.ElasticJobConfig;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@ElasticJobConfig(
        name = "com.base.core.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?",
        description = "可靠性投递消息补偿",
        overwrite = true,
        shardingTotalCount =1    //一个项目一张表，一个分片就够了，100张表用10个分片
)
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {

    //抽取数据
    @Autowired
    private MessageStoreService messageStoreService;

    //重发消息
    @Autowired
    private RabbitBroker rabbitBroker;

    private static  final int MAX_RETRY_COUNT=3;


    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> messages= messageStoreService.fetchTimeoutMessage4Retry(BrokerMessageStatus.SENDING);
        log.info("----------------@@@@@ 抓取数据集合，数量:{}---------------------",messages.size());
        return messages;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> dataList) {
        dataList.forEach(brokerMessage -> {
                    String messageId=brokerMessage.getMessageId();
                    if(brokerMessage.getTryCount()>=MAX_RETRY_COUNT){
                        this.messageStoreService.fail(messageId);
                        log.warn("---------------消息设置为最终失败，消息id:{}",messageId);
                    }else{
                        //每次重发需要更新一下try count字段
                        this.messageStoreService.updateTryCount(messageId);
                        //重发消息
                        this.rabbitBroker.reliantSend(brokerMessage.getMessage());
                    }
        });
    }
}
